package scala.org.zjt.spark

import org.apache.spark.{SparkConf, SparkContext}

/**
  * DESC  combineByKey 的聚合操作（类似reduceByKey和aggregate）
  *
  *   combineByKey  （ value初始化 -> 自增相加 -> 合并）
  *     1、参数1 初始化value的类型和值（b类）
  *     2、执行初始化后相同key的value对应类(b类)的自增相加
  *     3、执行自增后相同key的value对应类(b类)的合并
  *
  * @author
  * @create 2017-05-10 下午3:17
  **/
object PairRDDOption extends App {
  val sparkConfig = new SparkConf().setMaster("local").setAppName("PairRDDOption")
  val sparkContext = new SparkContext(sparkConfig)
  sparkContext.setLogLevel("ERROR")

  val initialScores = Array(("Fred", 98.1), ("Fred", 95.0), ("Fred", 91.0), ("Wilma", 93.0), ("Wilma", 95.0), ("Wilma", 98.0))
  val d1 = sparkContext.parallelize(initialScores)


  // TODO:  声明新的类型
  type score = (Double, Int)

  // TODO: 可实现多个聚合结果，类似于reduceByKey,对value操作，对value操作完后再加上name。
  val result = d1.combineByKey(
    value => (value, 1),        // TODO:  每个value初始化状态(总分数，个数)
    (sum: score, newScore: Double) => (sum._1 + newScore, sum._2 + 1),
    (sum1: score, sum2: score) => (sum1._1 + sum2._1, sum1._2 + sum2._2)
  ).filter{case (name,(value,num)) => num != 0 }.map{case (name,(value,num)) => (name,value / num) }.collect


  println(result.mkString(","))


}
